package qq2564874169.tuaotuao.fx.app.ro;

import org.apache.zookeeper.*;
import qq2564874169.tuaotuao.fx.FxAssert;
import qq2564874169.tuaotuao.fx.concurrent.FxCountDownLatch;
import qq2564874169.tuaotuao.fx.utils.ExUtil;

import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;

public class ZkHostProvider implements ROContext.HostProvider {
    private static String rootNode;

    public static void createRootNode(ZooKeeper zk) {
        createRootNode(zk, "/FxRO");
    }

    public static void createRootNode(ZooKeeper zk, String nodeName) {
        try {
            if (zk.exists(nodeName, false) == null) {
                zk.create(nodeName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            rootNode = nodeName;
        } catch (KeeperException | InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }
    private ZooKeeper zk;
    private String host;
    private Map<String, Hosts> hostMap = new Hashtable<>(8);

    public ZkHostProvider(ZooKeeper zk, String host) {
        FxAssert.isFalse(host.contains("#"), "host中不能包含#符号");
        this.zk = zk;
        this.host = host;
        try {
            init();
        } catch (InterruptedException | KeeperException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    private void init() throws InterruptedException, KeeperException {
        var latch = new FxCountDownLatch(1);
        zk.addWatch(rootNode, this::watchNode, AddWatchMode.PERSISTENT_RECURSIVE, (p1, p2, p3) -> latch.countDown(), null);
        latch.await();
        zk.getChildren(rootNode, false, (rc, path, ctx, children, stat) -> {
            try {
                for (var child : children) {
                    var hostNodes = zk.getChildren(path + "/" + child, false);
                    var hosts = new Hosts();
                    for (var h : hostNodes) {
                        hosts.add(h);
                    }
                    hostMap.put(child, hosts);
                }
            } catch (KeeperException | InterruptedException e) {
                e.printStackTrace();
            }
        }, null);
    }

    private void watchNode(WatchedEvent e) {
        if (e.getState().equals(Watcher.Event.KeeperState.Closed)) {
            return;
        }
        var path = e.getPath().substring(rootNode.length() + 1);
        var idx = path.indexOf("/");
        if (idx < 0) {
            return;
        }
        var ro = path.substring(0, idx);
        var host = path.substring(idx + 1);
        FxAssert.isFalse(host.contains("/"));
        if (e.getType().equals(Watcher.Event.EventType.NodeCreated)) {
            if (!hostMap.containsKey(ro)) {
                hostMap.put(ro, new Hosts());
            }
            hostMap.get(ro).add(host);
        } else if (e.getType().equals(Watcher.Event.EventType.NodeDeleted)) {
            if (hostMap.containsKey(ro)) {
                hostMap.get(ro).remove(host);
            }
        }
    }

    @Override
    public String get(String name, Class<?> type) {
        var hosts = hostMap.getOrDefault(name, null);
        if (hosts != null) {
            return hosts.get();
        }
        return null;
    }

    @Override
    public void register(String name, Class<?> type) {
        var node = rootNode + "/" + name;
        try {
            if (zk.exists(node, false) == null) {
                zk.create(node, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (InterruptedException e) {
            throw ExUtil.toRuntime(e);
        } catch (KeeperException e) {
            if (e.code() != KeeperException.Code.NODEEXISTS) {
                throw ExUtil.toRuntime(e);
            }
        }
        try {
            var hostNode = node + "/" + host.replaceAll("/", "#");
            zk.create(hostNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        } catch (InterruptedException e) {
            throw ExUtil.toRuntime(e);
        } catch (KeeperException e) {
            if (e.code() != KeeperException.Code.NODEEXISTS) {
                throw ExUtil.toRuntime(e);
            }
        }
    }

    private static class Hosts {
        private List<String> list = new ArrayList<>();
        private int index = 0;

        public synchronized void add(String host) {
            list.add(host.replaceAll("#", "/"));
        }

        public synchronized void remove(String host) {
            list.remove(host.replaceAll("#", "/"));
        }

        public synchronized String get() {
            if (list.isEmpty()) {
                return null;
            }
            if (index >= list.size()) {
                index = 0;
            }
            return list.get(index++);
        }
    }
}
